package com.dz.time;

import com.dz.stock.StockPrice;
import com.dz.stock.StockSource;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 * @ClassName AggregateFunctionExample
 * @Description TODO
 * @Author zhangdong
 * @Date 2020/11/24 13:36
 * @Version 1.0
 */
public class AggregateFunctionExample
{
    public static void main(String[] args) throws Exception {

         StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStream<StockPrice> stockStream = environment
                .addSource(new StockSource("D:\\git_code\\my\\hatchflink\\flink-advance\\src\\main\\resources\\stock/stock-tick-20200108.csv"));

        //根据symbol进行keyby   使用10s的滚动窗口
        DataStream<Tuple2<String, Double>> average = stockStream
                .keyBy(s -> s.symbol)
                .timeWindow(Time.seconds(10))
                .aggregate(new AverageAggregate());

        average.print();

        environment.execute("window aggregate function");
    }

    public static   class  AverageAggregate implements AggregateFunction<StockPrice, Tuple3<String, Double, Integer>, Tuple2<String, Double>>
    {

        @Override
        public Tuple3<String, Double, Integer> createAccumulator() {
            return Tuple3.of("",0d,0);
        }

        @Override
        public Tuple3<String, Double, Integer> add(StockPrice value, Tuple3<String, Double, Integer> accumulator) {
          //总量
           double current = value.price+accumulator.f1;
           //数量+1
           int count = accumulator.f2+1;
            return Tuple3.of(value.symbol,current,count);
        }

        @Override
        public Tuple2<String, Double> getResult(Tuple3<String, Double, Integer> accumulator) {

            //返回平均数 某个项 总金额数/总时刻数
            return Tuple2.of(accumulator.f0, accumulator.f1/accumulator.f2);
        }

        @Override
        public Tuple3<String, Double, Integer> merge(Tuple3<String, Double, Integer> a, Tuple3<String, Double, Integer> b) {
           //进行合并操作 类似reduce算子操作
            return  Tuple3.of(a.f0,a.f1+b.f1,a.f2+b.f2);
        }
    }
}
